package com.zyb.dfs.namenode.server;


import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.ListIterator;

/**
 * 负责管理edits log日志的核心组件
 *
 * @author zhonghuashishan
 */
public class FSEditlog {

    /**
     * editlog清理文件的时间间隔
     */
    private static final Long EDITLOG_CLEAN_INTEVAL = 30 * 1000L;

    /**
     * 元数据管理组件
     */
    private FSNamesystem namesystem;

    /**
     * 当前递增到的txid的序号
     */
    private long txidSeq = 0L;
    /**
     * 内存双缓冲区
     */
    private DoubleBuffer doubleBuffer = new DoubleBuffer();
    /**
     * 当前是否在将内存缓冲刷入磁盘中
     */
    private volatile Boolean isSyncRunning = false;

    /**
     * 在同步到磁盘中的最大的一个txid
     */
    private volatile Long syncTxid = 0L;
    /**
     * 是否正在调度刷盘的操作
     */
    private volatile Boolean isSchedulingSync = false;
    /**
     * 每个线程自己本地的txid副本
     */
    private ThreadLocal<Long> localTxid = new ThreadLocal<Long>();

    // 就会导致说，对一个共享的map数据结构出现多线程并发的读写的问题
    // 此时对这个map的读写是不是就需要加锁了
    // private Map<Thread, Long> txidMap = new HashMap<Thread, Long>();

    public FSEditlog(FSNamesystem namesystem) {
        this.namesystem = namesystem;
        EditLogCleaner editLogCleaner = new EditLogCleaner();
        editLogCleaner.start();
    }

    /**
     * 记录edits log日志
     */
    public void logEdit(String content) {
        // 这里必须得直接加锁
        synchronized (this) {
            // 刚进来及需要检查下是否有人在调度刷盘工作
            waitSchedulingSync();

            // 获取全局唯一递增的txid，代表了edits log的序号
            txidSeq++;
            long txid = txidSeq;

            // 放到ThreadLocal里去，相当于就是维护了一份本地线程的副本
            localTxid.set(txid);

            // 构造一条edits log对象
            EditLog log = new EditLog(txid, content);

            try {
                // 将edits log写入内存缓冲中，不是直接刷入磁盘文件
                doubleBuffer.write(log);
            } catch (Exception e) {
                e.getStackTrace();
            }

            // 每次写完一条数据之后，及需要检查当前缓冲区是否满了
            if (!doubleBuffer.shouldSyncToDisk()) {
                return;
            }

            // 如果代码进行这里说明需要刷磁盘
            isSchedulingSync = true;
        }

        // 释放掉锁

        logSync();
    }

    /**
     * 等待正在调度刷盘的操作
     */
    private void waitSchedulingSync() {
        try {
            while (isSchedulingSync) {
                // 此时就释放锁等待1s，在尝试获取锁
                wait(1000);
            }
        } catch (Exception e) {

        }
    }

    /**
     * 将内存缓冲中的数据刷入磁盘文件中
     * 在这里尝试允许某一个线程一次性将内存缓冲中的数据刷入磁盘文件中
     * 相当于实现一个批量将内存缓冲数据刷磁盘的过程
     */
    private void logSync() {
        // 再次尝试加锁
        synchronized (this) {
            long txid = localTxid.get();
            // 如果说当前正好有人在刷内存缓冲到磁盘中去
            if (isSyncRunning) {
                // 那么此时这里应该有一些逻辑判断

                // 假如说某个线程已经把txid = 1,2,3,4,5的edits log都从syncBuffer刷入磁盘了
                // 或者说此时正在刷入磁盘中
                // 此时syncMaxTxid = 5，代表的是正在输入磁盘的最大txid
                // 那么这个时候来一个线程，他对应的txid = 3，此时他是可以直接返回了
                // 就代表说肯定是他对应的edits log已经被别的线程在刷入磁盘了
                // 这个时候txid = 3的线程就不需要等待了
                // 如果已经有人在同步txid=30之前的数据
                if (txid <= syncTxid) {
                    return;
                }

                // 此时再来一个txid = 9的线程的话，那么他会发现说，已经有线程在等待刷下一批数据到磁盘了
                // 此时他会直接返回
                // 假如说此时来一个txid = 6的线程，那么的话，他是不好说的
                // 他就需要做一些等待，同时要释放掉锁

                try {
                    while (isSyncRunning) {
                        wait(2000);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }

            // 交换两块缓冲区
            doubleBuffer.setReadyToSync();
            // 然后可以保存一下当前要同步到磁盘中去的最大的txid
            // 此时editLogBuffer中的syncBuffer这块区域，交换完以后这里可能有多条数据
            // 而且他里面的edits log的txid一定是从小到大的
            // 此时要同步的txid = 6,7,8,9,10,11,12
            // syncMaxTxid = 12
            // 当前这块缓冲区写入的最大的txid，例如把txid=30为止之前所有的editlog都会刷入磁盘中
            syncTxid = txid;
            // 设置当前正在同步到磁盘的标志位
            isSyncRunning = true;

            isSchedulingSync = false;
            // 唤醒哪些换卡在waitSchedulingSync中while循环的线程
            notifyAll();
        }

        try {
            // 开始同步内存缓冲的数据到磁盘文件里去
            // 这个过程其实是比较慢，基本上肯定是毫秒级了，弄不好就要几十毫秒
            doubleBuffer.flush();
        } catch (Exception e) {
            e.getStackTrace();
        }

        synchronized (this) {
            // 同步完了磁盘之后，就会将标志位复位，再释放锁
            isSyncRunning = false;
            // 唤醒可能正在等待他同步完磁盘的线程
            notifyAll();
        }
    }

    /**
     * 获取已经刷入磁盘的editslog数据
     *
     * @return
     */
    public List<String> getFlushedTxids() {
        return doubleBuffer.getFlushedTxids();
    }

    /**
     * 强制把内存缓冲刷入到磁盘中
     */
    public void flush() {
        try {
            doubleBuffer.setReadyToSync();
            doubleBuffer.flush();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取当前缓冲区的数据
     *
     * @return
     */
    public String[] getBufferedEditsLog() {
        synchronized (this) {
            return doubleBuffer.getBufferedEditsLog();
        }
    }

    /**
     * 自动清理editlog文件
     */
    class EditLogCleaner extends Thread {
        @Override
        public void run() {
            System.out.println("editlog日志文件后台线程启动");
            while (true) {
                try {
                    Thread.sleep(EDITLOG_CLEAN_INTEVAL);
                    List<String> flushedTxids = getFlushedTxids();
                    if (flushedTxids != null && flushedTxids.size() > 0) {
                        long checkpointTxid = namesystem.getCheckpointTxid();

                        for (String flushedTxid : flushedTxids) {
                            String[] flushedTxidSplited = flushedTxid.split("_");
                            long startTxid = Long.valueOf(flushedTxidSplited[0]);
                            long endTxid = Long.valueOf(flushedTxidSplited[1]);
                            if (checkpointTxid >= endTxid) {
                                // 删除这个文件
                                File file = new File("/Users/admin/editslog/edits-" + startTxid + "-" + endTxid + ".log");
                                if (file.exists()) {
                                    file.delete();
                                }
                            }

                        }
                    }
                } catch (Exception e) {
                    e.getStackTrace();
                }
            }
        }
    }
}
